package com.fary.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.fary.bo.NotifyMerchantBO;
import com.fary.constant.RabbitMQExchangeConstant;
import com.fary.constant.RedisPrefixConstant;
import com.fary.entity.ConsumerLog;
import com.fary.impl.ConsumerLogServiceImpl;
import com.fary.service.PayMessageTemplateService;
import com.fary.utils.RedisCache;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;

import static com.fary.constant.RabbitMQExchangeConstant.WECHAT_TEMPLATE_EXCHANGE;
import static com.fary.constant.RabbitMQQueueConstant.WECHAT_TEMPLATE_QUEUE;

@Component
@Slf4j
@RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = WECHAT_TEMPLATE_QUEUE, durable = "true", autoDelete = "true", arguments = @Argument(name = "x-queue-mode", value = "lazy")),
        exchange = @Exchange(value = WECHAT_TEMPLATE_EXCHANGE)
))
public class WechatTemplateConsumer {
    @Autowired
    private RedisCache redisCache;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private PayMessageTemplateService payMessageTemplateService;
    @Autowired
    private ConsumerLogServiceImpl mayiktConsumerLogServiceImpl;
    /**
     * 最大重试次数
     */
    @Value("${spring.rabbitmq.listener.simple.retry.max-attempts}")
    private Integer maxAttempts;

    /**
     * 消费者订阅微信模板队列主题  调用 微信公众号模板接口
     */
    @RabbitHandler
    public void process(String msg, Message message, Channel channel) throws IOException {
        log.info(">>调用微信模板接口发送微信模板提醒:{}<<", msg);
        NotifyMerchantBO notifyMerchantBO = null;
        String msgId = null;
        try {
            notifyMerchantBO = JSONUtil.toBean(msg, NotifyMerchantBO.class);
            // 为了解决幂等性问题 提前根据消息id查询 消费者是否消费过
            msgId = notifyMerchantBO.getPaymentId();
            ConsumerLog mayiktConsumerLog = mayiktConsumerLogServiceImpl.getByMsgIdConsumerLog(msgId);
            if (mayiktConsumerLog != null) {
                // 已经消费过 手动ack告诉我们的mq删除该msg 避免重复消费
                return;
            }
            //1.调用第三方微信公众号模板接口
            //2.该方法返回值true 手动ack成功 返回false手动ack失败 需要MQ给我们消费者补偿
            boolean result = payMessageTemplateService.notifyMerchantPaymentResult(notifyMerchantBO);
            if (!result) {
                //返回false手动ack失败 需要MQ给我们消费者补偿
                //拒绝签收 抛出异常（mq服务器端间隔一段重试该msg）
                log.error("[调用消息模板接口发送消息失败,notifyMerchantBO:{}]", notifyMerchantBO);
                throw new RuntimeException("调用消息模板接口发送消息失败..");
            }

        } catch (Exception e) {
            log.error("e:{}", e);
            //手动ack失败 需要MQ给我们消费者补偿
            //拒绝签收 抛出异常（mq服务器端间隔一段重试该msg）
            //将该msg存放在死信队列中
            // 1.获取该msg被mq重试次数
            String redisSpareTireCount = redisCache.getCacheObject(RedisPrefixConstant.MSG_ID + msgId);
            int spareTireCount = StrUtil.isEmpty(redisSpareTireCount) ? 0 : Integer.parseInt(redisSpareTireCount);
            // 2.计数
            Integer newSpareTireCount = spareTireCount + 1;
            // 3.判断是否达到重试阈值 达到 将该msg转移存放死信队列中
            if (newSpareTireCount >= maxAttempts) {
                rabbitTemplate.convertAndSend(RabbitMQExchangeConstant.RETRY_EXCHANGE, "", msg);
                log.info("消费者消费该msg:{}多次失败，将msg存放死信队列中", msg);
                // 5.则需要清空redis对应的msgid 记录重试次数
                redisCache.deleteObject(RedisPrefixConstant.MSG_ID + msgId);
                return;
            }
            //4.将计数存放到redis
            redisCache.setCacheObject(RedisPrefixConstant.MSG_ID + msgId, newSpareTireCount);
            throw new RuntimeException(e);
        }
    }
}
